package jdklearning.concurrent.future;


import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

/**
 * @author shenenlu 2021年03月09日 下午16:01:39
 */
public class TimeCost {
    /**
     * 生产者，比作需要提前进行执行的操作
     * @return
     */
    private int provider() throws InterruptedException {
        System.out.println(String.format("provider time=%d", System.currentTimeMillis()));
        // 模拟耗时操作
        Thread.sleep(3000);
        return 1;
    }

    /**
     * 消费者，耗时操作
     */
    private int consumer(int message) throws InterruptedException {
        System.out.println(String.format("consumer time=%d",System.currentTimeMillis()));
        // 模拟耗时操作
        Thread.sleep(3000);
        System.out.println(String.format("consumer message=%s, time=%d" , message, System.currentTimeMillis()));
        return message * 2;
    }


    public void syncTest() throws InterruptedException {
        long startTime = System.currentTimeMillis();

        int oneProviderResult = provider();
        int twoProviderResult = provider();
        int oneConsumerResult = consumer(oneProviderResult);
        int twoConsumerResult = consumer(twoProviderResult);
        System.out.println(String.format("sync: one=%d, two=%d", oneConsumerResult, twoConsumerResult));
        long endTime = System.currentTimeMillis();
        System.out.println(String.format("sync time=%d", (endTime - startTime)));
    }


    /**
     * 此处采用了 Executors.newCachedThreadPool() 线程池，并采用默认配置。实际项目中，建议自定义线程池或者进行参数配置。
     * 在 consumer 中直接使用了全局变量（中间数据结果），建议实际项目中不采用这种写法。
     * 每一个任务流程：创建任务、提交任务、任务执行、获取任务结果
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public void futureTaskTest() throws InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();

        ExecutorService executor = Executors.newCachedThreadPool();
        // 创建 provider 任务
        FutureTask<Integer> oneFuture = new FutureTask<>(this::provider);
        FutureTask<Integer> twoFuture = new FutureTask<>(this::provider);
        // 提交 provider 任务
        executor.submit(oneFuture);
        executor.submit(twoFuture);
        // 获取中间数据
        Integer oneProviderResult = oneFuture.get();
        Integer twoProviderResult = twoFuture.get();

        // 创建 consumer 任务
        FutureTask<Integer> threeFuture = new FutureTask<>(() -> consumer(oneProviderResult));
        FutureTask<Integer> fourFuture = new FutureTask<>(() -> consumer(twoProviderResult));
        // 提交 consumer 任务
        executor.submit(threeFuture);
        executor.submit(fourFuture);
        // 获取结果
        Integer oneConsumerResult = threeFuture.get();
        Integer twoConsumerResult = fourFuture.get();

        System.out.println(String.format("sync: one=%d, two=%d", oneConsumerResult, twoConsumerResult));

        System.out.println(String.format("syncOther: time=%d", System.currentTimeMillis() - startTime));
    }


    /**
     * CompletableFuture 相较于 FutureTask ，此实践使用到的是 主动计算 的特性，
     * 多个异步任务之间依赖执行，自动通知下一个任务执行。
     *
     * CompletableFuture 相较于 FutureTask ，还使用到了多个任务编排，此实践使用的是 allOf()，即多个任务都必须执行完毕。
     * 较常用于最后进行等待。另外，还有比如 any 之类的编排 API 。
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void completableFutureTest() throws ExecutionException, InterruptedException {
        long startTime = System.currentTimeMillis();

        ExecutorService executor = Executors.newCachedThreadPool();
        CompletableFuture<Integer> one = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        return provider();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor)
                .thenApplyAsync((res) -> {
                    try {
                        return consumer(res);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor);

        CompletableFuture<Integer> two = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        return provider();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor)
                .thenApplyAsync((res) -> {
                    try {
                        return consumer(res);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor);

        /**
         * 等待执行，需要等待 one, two 两个任务均执行完毕
         */
        CompletableFuture.allOf(one, two);
        System.out.println(String.format("sync: one=%d, two=%d", one.get(), two.get()));
        System.out.println(String.format("syncOther: time=%d", System.currentTimeMillis() - startTime));
    }


    public void completableFutureTest_01() throws ExecutionException, InterruptedException {
        long startTime = System.currentTimeMillis();

        ExecutorService executor = Executors.newCachedThreadPool();
        CompletableFuture<Integer> one = completableFutureExecute(executor);
        CompletableFuture<Integer> two = completableFutureExecute(executor);
        /**
         * 等待执行，需要等待 one, two 两个任务均执行完毕
         */
        CompletableFuture.allOf(one, two);
        System.out.println(String.format("sync: one=%d, two=%d", one.get(), two.get()));

        System.out.println(String.format("syncOther: time=%d", System.currentTimeMillis() - startTime));
    }




    // 抽象
    private CompletableFuture<Integer> completableFutureExecute(ExecutorService executor) {
        return CompletableFuture
                .supplyAsync(() -> {
                    try {
                        return provider();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor)
                .thenApplyAsync((res) -> {
                    try {
                        return consumer(res);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        return 0;
                    }
                }, executor);
    }



    public static void main(String[] args) throws InterruptedException, ExecutionException {
        TimeCost timeCost = new TimeCost();
//        timeCost.syncTest();
        timeCost.futureTaskTest();
    }
}
